package com.fengye.revoker;

import com.alibaba.fastjson.JSON;
import com.fengye.config.PropertyConfigHelper;
import com.fengye.model.ProviderService;
import com.fengye.model.RemoteResponse;
import com.fengye.serialization.NettyDecodeHandler;
import com.fengye.serialization.NettyEncodeHandler;
import com.fengye.serialization.common.SerializeType;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CountDownLatch;

/**
 * @Author fengye
 * @Date 2023/2/14 15:46
 **/
public class NettyChannelPoolFactory {
    private final static Logger logger = LoggerFactory.getLogger(NettyChannelPoolFactory.class);
    /** 单例工厂 */
    private final static NettyChannelPoolFactory channelPoolFactory = new NettyChannelPoolFactory();

    /** Key为服务提供者地址，value为Netty Channel阻塞队列 */
    private final static Map<InetSocketAddress, ArrayBlockingQueue<Channel>> channelPoolMap = Maps.newConcurrentMap();

    /** 初始化Netty Channel阻塞队列的长度 */
    private final static int channelConnectSize = PropertyConfigHelper.getChannelConnectSize();
    /** 初始化序列化协议类型，该值为可配置（这个现在是无法给调用方选择的，做成可选？） */
    private final static SerializeType serializeType = PropertyConfigHelper.getSerializeType();
    /** 服务提供者列表 */
    private List<ProviderService> serviceMetaDataList = Lists.newArrayList();

    private NettyChannelPoolFactory() { }

    public static NettyChannelPoolFactory channelPoolFactoryInstance() {
        return channelPoolFactory;
    }

    /**
     * 初始化Netty channel 连接队列Map
     * @param providerMap
     */
    public void initChannelPoolFactory(Map<String, List<ProviderService>> providerMap) {
        // 将服务提供者信息存入serviceMetaDataList
        Collection<List<ProviderService>> collectionServiceMetaDataList = providerMap.values();
        for(List<ProviderService> serviceMetaDataModels : collectionServiceMetaDataList) {
            if(CollectionUtils.isEmpty(serviceMetaDataModels)) {
                continue;
            }
            serviceMetaDataList.addAll(serviceMetaDataModels);
        }

        // 获取服务提供者地址列表
        Set<InetSocketAddress> socketAddressSet = Sets.newHashSet();
        for(ProviderService serviceMetaData : serviceMetaDataList) {
            String serviceIp = serviceMetaData.getServerIp();
            int servicePort = serviceMetaData.getServerPort();

            InetSocketAddress socketAddress = new InetSocketAddress(serviceIp, servicePort);
            socketAddressSet.add(socketAddress);
        }

        for(InetSocketAddress socketAddress:socketAddressSet) {
            int realChannelConnectSize = 0;
            while (realChannelConnectSize < channelConnectSize) {
                Channel channel = null;
                while (channel == null) {
                    // 若channel不存在，则注册新的Netty Channel, 如果一直连接不上一直循环？
                    channel =registerChannel(socketAddress);
                    logger.info("connect channel info:" + JSON.toJSONString(channel));
                }
                // 计数器，初始化的时候存入阻塞队列的Netty channel个数不超过channelConnectSize
                realChannelConnectSize++;

                ArrayBlockingQueue<Channel> channelArrayBlockingQueue = channelPoolMap.get(socketAddress);
                if (channelArrayBlockingQueue == null) {
                    channelArrayBlockingQueue = new ArrayBlockingQueue<Channel>(channelConnectSize);
                    channelPoolMap.put(socketAddress, channelArrayBlockingQueue);
                }
                // 放入阻塞队列尾
                channelArrayBlockingQueue.offer(channel);
            }
        }

    }

    /**
     * 根据服务提供者地址获取对应的Netty Channel阻塞队列
     *
     * @param socketAddress
     * @return
     */
    public ArrayBlockingQueue<Channel> acquire(InetSocketAddress socketAddress) {
        return channelPoolMap.get(socketAddress);
    }

    /**
     * Channel使用完后，回收到堵塞队列中
     * @param arrayBlockingQueue
     * @param channel
     * @param socketAddress
     */
    public void release(ArrayBlockingQueue<Channel> arrayBlockingQueue, Channel channel, InetSocketAddress socketAddress) {
        if(arrayBlockingQueue == null) {
            return;
        }

        // 回收之前先检查channel是否可用，不可用的话，把不可用的关闭，然后重新注册一个，放入阻塞队列，
        if(channel == null || !channel.isActive() || !channel.isOpen() || !channel.isWritable()) {
            if(channel != null) {
                channel.deregister().syncUninterruptibly().awaitUninterruptibly();
                channel.closeFuture().syncUninterruptibly().awaitUninterruptibly();
            }
            Channel newChannel = null;
            while (newChannel == null) {
                newChannel = registerChannel(socketAddress);
                logger.info("connect channel info:" + JSON.toJSONString(newChannel));
            }
            arrayBlockingQueue.offer(newChannel);
            return;
        }
        arrayBlockingQueue.offer(channel);
    }

    /**
     * 与服务提供这建立连接Channel
     * @param socketAddress
     * @return
     */
    public Channel registerChannel(InetSocketAddress socketAddress) {
        // 这是一个线程池，同一时间一个线程EventLoop一个channel， 设置10个线程
        EventLoopGroup group = new NioEventLoopGroup(10);
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.remoteAddress(socketAddress);

        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        // 注册Netty编码器
                        socketChannel.pipeline().addLast(new NettyEncodeHandler(serializeType));
                        socketChannel.pipeline().addLast(new NettyDecodeHandler(RemoteResponse.class, serializeType));
                        socketChannel.pipeline().addLast(new NettyClientInvokeHandler());
                    }
                });

        try {
            ChannelFuture channelFuture = bootstrap.connect().sync();
            final Channel newChannel = channelFuture.channel();
            final CountDownLatch connectedLatch = new CountDownLatch(1);

            final List<Boolean> isSuccessHolder = Lists.newArrayListWithCapacity(1);
            // 监听Channel是否建立成功
            channelFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture channelFuture) throws Exception {
                    // 如果Channel建立成功，保存建立成功的标记
                    if(channelFuture.isSuccess()) {
                        isSuccessHolder.add(Boolean.TRUE);
                    } else {
                        channelFuture.cause().printStackTrace();
                        isSuccessHolder.add(Boolean.FALSE);
                    }
                    connectedLatch.countDown();
                }
            });

            connectedLatch.wait();

            if (isSuccessHolder.get(0)) {
                return newChannel;
            }
        } catch (InterruptedException e) {
            logger.error("connect new channel is error", e);
            throw new RuntimeException(e);
        }
        return null;
    }
}
